package com.example.reactor;

import org.reactivestreams.Subscription;
import reactor.core.Disposable;
import reactor.core.publisher.BaseSubscriber;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;
import reactor.core.publisher.SignalType;

import java.io.IOException;
import java.time.Duration;

/**
 * Mono<T>： 代表0或1个元素的异步序列。它用于处理单个结果（或空结果/错误）的场景。
 * 示例：根据ID查询一个用户、执行一个返回void/状态的更新操作、获取一个HTTP请求的响应体（通常是一个JSON对象）。
 * Flux<T>： 代表0到N个（0个、1个或多个）元素的异步序列。它用于处理多个结果的流式场景。
 * 示例：查询所有用户列表、监听实时事件流（如消息队列、SSE）、读取一个大文件的分块内容。
 * ---------
 * 当说 Mono 或 Flux 可以表示 0个元素时，意味着它们代表一个没有实际数据元素的异步序列，但会发出明确的完成信号（onComplete）（或错误信号 onError），而不是无限等待或什么都不做。
 */
public class FluxDemo1 {

    /**
     * 流的信号：SignalType
     *
     */

    /**
     * 第一个 doOnNext() 只能感知 Flux.just() 发出的原始数据，而第二个 doOnNext() 能感知 map() 转换后的数据。
     * 异常处理隔离：每个操作符独立处理自己的异常。例如 map() 中发生的异常不会被其上游的 doOnError() 捕获，但会被下游的 doOnError() 捕获。
     * 订阅触发执行：只有在调用 subscribe() 后，整个操作符链才会从下游向上游依次触发执行，但每个操作符只能感知其直接上游的数据和事件。
     * 因此，方法的感知是单向且有限的，只能感知其上游的操作结果，无法感知下游或其他分支的操作。
     *
     * @param args
     */
    public static void main(String[] args) {
        Flux.range(1, 7)
                .map(integer -> 10 / integer)
                .subscribe(System.out::println);
        System.out.println("------------------");
        /**
         * doOnXxx：感知某个流事件，写在该流的后面
         * doOnNext：每个数据（流的数据）到达的时候触发
         * doOnEach：每个数据（流数据->流的数据和信号）到达的时候触发
         */
        Flux.just(1, 2, 3, 4, 5, 0, 6, 7, 8, 9)
                /**
                 * 仅感知just方法，不会感知map方法
                 */
                .doOnNext(integer -> {
                    System.out.println("OnNext：" + integer);
                })
                /**
                 * 感知doOnNext方法
                 * doOnEach方法比doOnNext方法更通用，可以处理任意的信号类型，如 onComplete、onError、onSubscribe 等。
                 */
                .doOnEach(signal -> {
                    System.out.println("OnEach：" + signal.get());
                })
                .map(integer -> 10 / integer)
                /**
                 * 感知map方法
                 */
                .doOnNext(integer -> {
                    System.out.println("OnNext：" + integer);
                })
                .doOnError(throwable -> {
                    System.out.println("异常：" + throwable);
                })
                .doFinally(signalType -> {
                    System.out.println("完成：" + signalType);
                })
                .subscribe(System.out::println);
    }

    /**
     * 测试Mono
     */
    public void fluxDoOn(String[] args) throws InterruptedException {
//        Mono<Integer> just = Mono.just(1);
//        just.subscribe(e -> System.out.println("接收数据：" + e));

        /**
         * 创建一个空的Mono，表示一个没有实际数据的异步序列，但会发出一个完成信号。
         */
        Mono<Integer> emptyMono = Mono.empty();
        emptyMono.subscribe(e -> System.out.println("接收数据：" + e));

        /**
         * 创建一个空的Flux，表示一个没有实际数据的异步序列，但会发出一个完成信号。
         */
        Flux<Integer> flux = Flux.range(1, 9)
                /**
                 * 延迟1秒发送数据
                 * 下面的流操作上面的流
                 */
                .delayElements(Duration.ofSeconds(1))
                /**
                 * 事件感知：当流发生什么事件时，会触发相应的回调方法。（钩子函数）
                 */
                .doOnComplete(() -> {
                    System.out.println("完成");
                })
                .doOnCancel(() -> {
                    System.out.println("取消");
                })
                .doOnError(throwable -> {
                    System.out.println("异常：" + throwable);
                })
                .doOnNext(e -> {
                    if (e == 2) {
                        int i = 1 / 0;
                    }
                    System.out.println("OnNext：" + e);
                })
                .doFinally(signalType -> {
                    System.out.println("完成：" + signalType);
                });
        /**
         * 流不订阅，则不会产生作用
         * 直接在flux.subscribe();中引入订阅者
         */
        flux.subscribe(new BaseSubscriber<Integer>() {
            /**
             * 钩子函数：订阅者订阅流时，会触发相应的回调方法。
             */
            @Override
            protected void hookOnSubscribe(Subscription subscription) {
                System.out.println("订阅关系" + subscription);
                /**
                 * 订阅者向发布者请求数据：背压
                 * 触发回调函数onNext
                 */
                request(1);
            }

            /**
             * 由doOnNext()触发，触发回调函数hookOnNext
             * 先执行doONNext()，再执行hookOnNext()
             * @param value the emitted value to process
             */
            @Override
            protected void hookOnNext(Integer value) {
                if (value > 5) {
                    cancel();
                } else {
                    if (value == 3) {
//                        throw new RuntimeException("嘿嘿，你有异常");
                        int i = 1 / 0;
                    }
                    System.out.println("接收数据：" + value);
                    request(1);
                }
            }

            @Override
            protected void hookOnComplete() {
                System.out.println("数据流完成");
            }

            /**
             * 感知的是订阅者的异常，不是发布者的异常
             * 订阅者的错误，不是发布者的错误，发布者只是单纯发送数据
             * @param throwable the error to process
             */
            @Override
            protected void hookOnError(Throwable throwable) {
                System.out.println("数据流异常：" + throwable);
            }

            @Override
            protected void hookOnCancel() {
                System.out.println("数据流取消");
            }

            @Override
            protected void hookFinally(SignalType type) {
                System.out.println("数据流完成：" + type);
            }
        });

        Thread.sleep(2000);

        /**
         * 原始流
         */
        Flux<Integer> range = Flux.range(1, 9);
        /**
         * 新流1
         */
        Flux<Integer> filter1 = range.filter(e -> e > 2);
        /**
         * 新流2
         */
        Flux<Integer> filter2 = range.filter(e -> e > 5);


        try {
            System.in.read();
        } catch (IOException e) {
            throw new RuntimeException(e);
        }
    }

    /**
     * 测试Flux
     */
    public void flux() {
        /**
         * Mono：0-1个数据
         * Flux：0-N个数据
         */

        /**
         * 创建发布者
         * 流不使用就无用，需要订阅
         */
        Flux<Integer> just = Flux.just(1, 2, 3, 4, 5);

        /**
         * 订阅者
         */
        just.subscribe(e -> System.out.println("接收数据：" + e));
        /**
         * 可以有多个订阅者
         * 对于每个消费者来说流都是一样的
         */
        just.subscribe(e -> System.out.println("接收数据，嘿嘿：" + e));

        System.out.println("结束");

        /**
         * 每秒产生一个从0开始递增的数字
         */
        Flux<Long> interval = Flux.interval(Duration.ofSeconds(1));
        interval.subscribe(e -> System.out.println("接收数据：" + e));


        try {
            System.in.read();
        } catch (IOException e) {
            System.out.println(e.getMessage());
        }
    }

}
